/*
 * Copyright 2011-2012 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.sabayframework.mem.impl.ram;

import it.unimi.dsi.fastutil.longs.Long2ObjectOpenHashMap;
import it.unimi.dsi.fastutil.longs.LongIterator;

import java.util.concurrent.locks.ReadWriteLock;

import org.sabayframework.mem.BigBuffer;
import org.sabayframework.mem.Memorizable;
import org.sabayframework.mem.Transaction.Status;
import org.sabayframework.mem.impl.BigBufferAllocator;
import org.sabayframework.mem.impl.TransactionImpl;
import org.sabayframework.mem.impl.TxActivity;
import org.sabayframework.mem.impl.ram.RAMSpace.Version;

class PessimisticTxActivity extends TxActivity{
	
	final BigBufferAllocator allocator;
	final Long2ObjectOpenHashMap<Version> records;	
	final ReadWriteLock lock;

	private final Long2ObjectOpenHashMap<Memorizable> rCache = new Long2ObjectOpenHashMap<Memorizable>();		
	private final Long2ObjectOpenHashMap<Memorizable> wCache = new Long2ObjectOpenHashMap<Memorizable>();		
	
	public PessimisticTxActivity(RAMSpace space, TransactionImpl transaction) {
		super(transaction, space);
		this.allocator = space.allocator;
		this.records = space.records;
		this.lock = space.lock;		
	}
	@Override
	protected Boolean[] contains(long[] ids) throws Exception {
		Boolean[] res = new Boolean[ids.length];
		boolean done = true;
		for(int i=0;i<ids.length;++i) {
			final long id = ids[i];
			Memorizable m = wCache.get(id);
			if(m == null && wCache.containsKey(id))res[i] = Boolean.FALSE;
			else{
				if(m == null) m = rCache.get(id);
				if(m == null) done = false;
				else res[i] = Boolean.TRUE;
			}
		}		
		if(done) return res;
		if(lock.readLock().tryLock()){
			try{
				for(int i=0;i<ids.length;++i) {
					if(res[i]!=null) continue;
					final long id = ids[i];
					Version v = records.get(id);
					if(v != null && v.data!=null) res[i] = Boolean.TRUE;
					else res[i] = Boolean.FALSE;
				}					
				return res;
			}finally{
				lock.readLock().unlock();
			}
		}
		return null;
	}

	@Override
	protected Memorizable[] read(Class<? extends Memorizable> objectClass, long[] ids) throws Exception {
		Memorizable[] res= new Memorizable[ids.length];
		boolean done = true;
		for(int i=0;i<ids.length;++i) {
			final long id = ids[i];
			Memorizable m = wCache.get(id);
			if(m == null && wCache.containsKey(id)) {
				m = objectClass.newInstance();
				m.setId(id);
			}
			if(m == null) m = rCache.get(id);
			if(m == null) done = false;
			res[i] = m;
		}		
		if(done) return res;
		if(lock.writeLock().tryLock()){
			try{
				final long txid = getTransaction().getId(); 
				//check all not locked
				for(int i=0;i<ids.length;++i) {
					if(res[i]!=null) continue;
					final long id = ids[i];
					Version v = records.get(id);
					if(v!=null && v.locked!=0 && v.locked!=txid) return null;// locked
				}
				for(int i=0;i<ids.length;++i) {
					if(res[i]!=null) continue;
					final long id = ids[i];
					Version v = records.get(id);
					BigBuffer data =null;
					if(v != null && v.data!=null) data = v.data;
					Memorizable m = objectClass.newInstance();
					m.setId(id);
					if(data!=null){
						data.rewind();
						m.deserialize(data);
					}
					if(v == null){
						v = new RAMSpace.Version();
						records.put(id, v);
					}
					v.locked = txid;
					rCache.put(id, m);
					res[i] = m;
				}					
				return res;
			}finally{
				lock.writeLock().unlock();
			}
		}
		return null;
	}

	@Override
	protected boolean write(Memorizable[] rs) throws Exception {
		boolean done = true;
		for(Memorizable r:rs){
			long id = r.getId();
			if(wCache.containsKey(id) || rCache.containsKey(id)){
				wCache.put(id, r);
				rCache.remove(id);
			}else{
				done = false;
				break;
			}
		}
		if(done) return true;
		if(lock.writeLock().tryLock()){
			try{
				final long txid = getTransaction().getId(); 
				//check all not locked
				for(Memorizable m : rs) {
					final long id = m.getId();
					Version v = records.get(id);
					if(v == null) continue;
					if(v.locked!=0 && v.locked!=txid) return false;// locked
				}
				for(Memorizable m : rs) {
					final long id = m.getId();
					Version v = records.get(id);
					if(v == null){
						v = new RAMSpace.Version();
						records.put(id, v);
					}
					v.locked = txid;
					wCache.put(id, m);
					rCache.remove(id);
				}					
				return true;
			}finally{
				lock.writeLock().unlock();
			}
		}		
		return true;
	}

	@Override
	protected boolean delete(long[] ids) throws Exception {
		boolean done = true;
		for(int i=0;i<ids.length;++i){
			long id = ids[i];
			if(wCache.containsKey(id) || rCache.containsKey(id)){
				wCache.put(id, null);
				rCache.remove(id);
			}else{
				done = false;
				break;
			}
		}
		if(done) return true;
		if(lock.writeLock().tryLock()){
			try{
				final long txid = getTransaction().getId(); 
				//check all not locked
				for(int i=0;i<ids.length;++i) {
					final long id = ids[i];
					Version v = records.get(id);
					if(v == null) continue;
					if(v.locked!=0 && v.locked!=txid) return false;// locked
				}
				for(int i=0;i<ids.length;++i) {
					final long id = ids[i];
					Version v = records.get(id);
					if(v == null){
						v = new RAMSpace.Version();
						records.put(id, v);
					}
					v.locked = txid;
					wCache.put(id, null);
					rCache.remove(id);
				}					
				return true;
			}finally{
				lock.writeLock().unlock();
			}
		}		
		return true;
	}

	
	private boolean commitPessimistic(){
		if(lock.writeLock().tryLock()){
			try{
				final long txid = getTransaction().getId(); 
				BigBuffer buffer = null;
				LongIterator it = wCache.keySet().iterator();
				while(it.hasNext()){
					long address = it.nextLong();
					Memorizable data = wCache.get(address);
					Version v = records.get(address);
					if(data == null) {
						//deleted 
						if(v!=null){
							v.lastModified = txid;
							if(v.data!=null) allocator.free(v.data);								
							v.data = null;
							v.locked = 0;
						}
					}else{
						final int maxLength = data.getMaxBufferSize();
						if(buffer == null || buffer.capacity() < maxLength){
							buffer = getSpace().wrap(new byte[maxLength], 0, maxLength);
						}
						buffer.clear();
						data.serialize(buffer);
						buffer.flip();
						if(v == null){
							//new record
							v = new RAMSpace.Version();
							v.lastModified = txid;
							v.data = allocator.allocate(buffer);
							records.put(address, v);
						}else{
							if(v.data!=null) allocator.free(v.data);								
							v.data = allocator.allocate(buffer);
							v.lastModified = txid;
							v.locked = 0;
						}
						
					}
				}
				it = rCache.keySet().iterator();
				while(it.hasNext()){
					long addr = it.nextLong();
					Version v = records.get(addr);
					if(v!=null)  v.locked = 0;
				}
				rCache.clear();
				wCache.clear();
				return true;
			} finally{
				lock.writeLock().unlock();
			}
		}
		return false;
	}
	private boolean rollbackPessimistic(){
		if(lock.writeLock().tryLock()){
			try{
				LongIterator it = rCache.keySet().iterator();
				while(it.hasNext()){
					long addr = it.nextLong();
					Version v = records.get(addr);
					if(v!=null)  v.locked = 0;
				}
				it = wCache.keySet().iterator();
				while(it.hasNext()){
					long addr = it.nextLong();
					Version v = records.get(addr);
					if(v!=null)  v.locked = 0;
				}
				rCache.clear();
				wCache.clear();
				return true;
			} finally{
				lock.writeLock().unlock();
			}
		}
		return false;
	}

	@Override
	protected boolean complete(final Status status) throws Exception {
		if(status == Status.COMMIT) return commitPessimistic();
		return rollbackPessimistic();
	}
}